package cn.juque.lucenecandy.core.datasync;

import cn.hutool.core.bean.BeanUtil;
import cn.juque.common.base.BaseResponseVO;
import cn.juque.common.constants.MessageEnum;
import cn.juque.common.exception.AppException;
import cn.juque.common.fastjson.abstracts.AbstractTypeReference;
import cn.juque.lucenecandy.bo.IndexUpdateParamBO;
import cn.juque.lucenecandy.cache.IpCache;
import cn.juque.lucenecandy.core.annotation.DataSync;
import cn.juque.lucenecandy.core.constants.UrlConstant;
import cn.juque.lucenecandy.core.datasync.listener.CommitListenerRender;
import cn.juque.lucenecandy.core.datasync.tcc.ITccService;
import cn.juque.lucenecandy.core.utils.SignatureUtil;
import cn.juque.lucenecandy.vo.IndexUpdateParamVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.validation.constraints.NotNull;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author juque
 * @version 1.0.0
 * <ul>
 *     <li>TccDataSyncService</li>
 * </ul>
 * @date 2023-12-29
 **/
@Slf4j
@Service("tccDataSyncService")
@DataSync(value = "tcc")
public class TccDataSyncServiceImpl implements IDataSyncService {

    @Resource
    private IpCache ipCache;

    @Resource
    private CommitListenerRender commitListenerRender;

    @Resource(name = "candyTryCommitService")
    private ITccService candyTryCommitService;

    @Resource(name = "candyCommitService")
    private ITccService candyCommitService;

    @Resource(name = "candyCancelService")
    private ITccService candyCancelService;

    /**
     * 更新
     *
     * @param param 数据
     */
    @Override
    public void commit(@NotNull IndexUpdateParamBO param) {
        // 预提交
        boolean flag = this.tryOperate(param);
        // 实际没有对其他节点执行预提交 或 提交成功
        if (flag) {
            this.commitOperate(param);
        } else {
            // 取消
            this.cancel(param);
            throw new AppException(MessageEnum.SYSTEM_ERROR);
        }
    }

    /**
     * 删除
     *
     * @param param 数据
     */
    @Override
    public void cancel(@NotNull IndexUpdateParamBO param) {
        switch (param.getSyncType()) {
            case ADD:
                this.candyCancelService.add(param.getContent());
                break;
            case UPDATE:
                this.candyCancelService.update(param.getContent());
                break;
            case DEL:
                this.candyCancelService.del(param.getClassName(), param.getIdList());
                break;
            default:
                break;
        }
        // cancel
        this.batchPost(param, SyncTypeEnum.CANCEL);
    }

    private boolean tryOperate(IndexUpdateParamBO param) {
        switch (param.getSyncType()) {
            case ADD:
                this.candyTryCommitService.add(param.getContent());
                break;
            case UPDATE:
                this.candyTryCommitService.update(param.getContent());
                break;
            case DEL:
                this.candyTryCommitService.del(param.getClassName(), param.getIdList());
                break;
            default:
                break;
        }
        return this.batchPost(param, SyncTypeEnum.PRE);
    }

    private void commitOperate(IndexUpdateParamBO param) {
        switch (param.getSyncType()) {
            case ADD:
                this.candyCommitService.add(param.getContent());
                break;
            case UPDATE:
                this.candyCommitService.update(param.getContent());
                break;
            case DEL:
                this.candyCommitService.del(param.getClassName(), param.getIdList());
                break;
            default:
                break;
        }
        // commit
        this.batchPost(param, SyncTypeEnum.COMMIT);
    }

    private boolean batchPost(IndexUpdateParamBO paramBO, SyncTypeEnum syncType) {
        // 事前操作
        if (!this.commitListenerRender.before(paramBO)) {
            return true;
        }
        String urlPath = null;
        switch (syncType) {
            case PRE:
                urlPath = UrlConstant.PRE_COMMIT_DOC;
                break;
            case COMMIT:
                urlPath = UrlConstant.COMMIT_DOC;
                break;
            case CANCEL:
                urlPath = UrlConstant.CANCEL_DOC;
                break;
            default:
                break;
        }
        IndexUpdateParamVO vo = new IndexUpdateParamVO();
        BeanUtil.copyProperties(paramBO, vo);
        List<String> ips = this.ipCache.listValidIp(false);
        try {
            Map<String, BaseResponseVO<Boolean>> resp = this.ipCache.batchPost(ips,
                    SignatureUtil.buildUrl(urlPath),
                    new AbstractTypeReference<BaseResponseVO<Boolean>>() {
                    }, vo);
            AtomicBoolean flag = new AtomicBoolean(true);
            resp.forEach((k, v) -> {
                if (MessageEnum.OK.getCode().equals(v.getCode()) && Boolean.TRUE.equals(v.getData())) {
                    return;
                }
                flag.set(false);
                log.error("ip:{} tcc post error, errorCode={}, errorMsg={}", k, v.getCode(), v.getMsg());
            });
            // 事后
            this.commitListenerRender.after(paramBO);
            return flag.get();
        } catch (Exception e) {
            log.error("batch post error", e);
            return false;
        }
    }
}
